package cn.hc.case2;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author huangchao E-mail:fengquan8866@163.com
 * @version 创建时间：2024/10/19 10:58
 */
public class DistributeLock {

    // 注意：逗号左右不能有空格
    private final String connectString = "master:2181,hadoop141:2181,hadoop-test:2181";
    private final int sessionTimeout = 2000;

    private final ZooKeeper zkClient;

    private final CountDownLatch connectLatch = new CountDownLatch(1);
    private final CountDownLatch waitLatch = new CountDownLatch(1);
    private String preNode;
    private String currentMode;

    public DistributeLock() throws IOException, InterruptedException, KeeperException {
        // 获取连接
        zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("-----------------------------------");
                // connectLatch 如果连接上zk 可以释放
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("connectLatch.countDown: " + System.currentTimeMillis());
                    connectLatch.countDown();
                }

                // waitLatch 需要释放
                if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(preNode)) {
                    waitLatch.countDown();
                }
            }
        });

        System.out.println("connectLatch.await: " + System.currentTimeMillis());
        // 等待zk正常连接后，往下走程序
        connectLatch.await();

        // 判断根节点/locks是否存在
        Stat stat = zkClient.exists("/hc/locks", false);
        if (stat == null) {
            // 创建一下根节点
            zkClient.create("/hc/locks", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }

    // 对zk加锁
    public void zklock() {
        // 创建对应的临时带序号节点
        try {
            currentMode = zkClient.create("/hc/locks/seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 判断创建的节点是否是最小的序号节点，如果是 获取到锁；如果不是，监听他序号前一个节点
            List<String> children = zkClient.getChildren("/hc/locks", false);
            System.out.println("children: " + children);
            // 如果children 只有一个值，那就直接获取锁；如果有多个节点，需要判断，谁最小
            if (children.size() == 1) {
                return;
            }
            Collections.sort(children);
            // 获取节点名称 seq-0000000000
            String thisNode = currentMode.substring("/hc/locks/".length());
            // 通过seq-0000000000 找到该节点在children集合的位置
            int index = children.indexOf(thisNode);

            // 判断
            if (index == -1) {
                System.out.println("数据异常");
            } else if (index == 0) {
                // 就一个节点，可以获取到锁了
                return;
            } else {
                // 需要监听 前一个节点变化
                preNode = "/hc/locks/" + children.get(index - 1);
                System.out.println("preNode: " + preNode);
                zkClient.getData(preNode, true, new Stat());

                // 等待监听
                waitLatch.await();
                return;
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    // 释放锁
    public void unZkLock() {
        // 删除节点
        try {
            System.out.println("unZkLock: " + this.currentMode);
            zkClient.delete(this.currentMode, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
}
